package pubsub

import (
	"encoding/hex"
	"errors"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	pbcom "hundsun.com/hsl/hschain/protos/common"
	"hundsun.com/hsl/hschain/protos/ledger"
	dbm "hundsun.com/hsl/hschain/store/db"
	"hundsun.com/hsl/hschain/store/mq"
)

// sub type
const (
	// SubBlock sub block
	SubBlock = iota
	// SubBlockHeader sub block header
	SubBlockHeader
	// SubBlockExecResult sub block exec result
	SubBlockExecResult
	// SubSysParam sub system param
	SubSysParam
)

// sub type--->name
var subName = map[int32]string{
	SubBlock:           "SubBlock",
	SubBlockHeader:     "SubBlockHeader",
	SubBlockExecResult: "SubBlockExecResult",
	SubSysParam:        "SubSysParam",
}

var (
	// ErrNotAllowSubscribeType error not allow subscribe type
	ErrNotAllowSubscribeType = errors.New("ErrNotAllowSubscribeType")
)

// wrapSub wrap subscribe
type wrapSub struct {
	ID        string // ID 用于区分订阅者
	subscribe *pbcom.ReqSubscriber
}

// PubSub pubsub manage
type PubSub struct {
	log         logging.Log
	ledgerDB    dbm.Database        // 存储账本数据句柄
	topicClient mq.Client           // topic类型client
	workClient  mq.Client           // work类型client
	systemSubs  map[string]*wrapSub // 系统参数订阅,即订阅执行合约作用于模块的参数
	normalSubs  map[string]*wrapSub // 常规数据订阅
	wg          *sync.WaitGroup
	mut         *sync.RWMutex
}

// NewPubSub new pubsub manage
func NewPubSub(log logging.Log, ldb dbm.Database, topicClient mq.Client, workClient mq.Client) *PubSub {
	service := &PubSub{
		log:         log,
		ledgerDB:    ldb,
		topicClient: topicClient,
		workClient:  workClient,
		systemSubs:  make(map[string]*wrapSub),
		normalSubs:  make(map[string]*wrapSub),
		wg:          &sync.WaitGroup{},
		mut:         &sync.RWMutex{},
	}
	service.init()
	return service
}

// 初始化: 从数据库读出subscriber的数目
func (pm *PubSub) init() {
	subscribes, err := pm.ListSubscriber()
	if err != nil || len(subscribes) == 0 {
		pm.log.Errorf("Push init %v", err)
		return
	}
	for _, sub := range subscribes {
		id := genSubscriberID(sub)
		sysd := sub.GetSystem()
		if sysd != nil {
			pm.systemSubs[id] = &wrapSub{subscribe: sub}
		} else {
			pm.normalSubs[id] = &wrapSub{subscribe: sub}
		}
	}
}

//ListSubscriber 列出所有已经设置的推送订阅
func (pm *PubSub) ListSubscriber() ([]*pbcom.ReqSubscriber, error) {
	pdb := NewSubscriber(pm.ledgerDB)
	return pdb.GetEntireSubscribers()
}

// Close ...
func (pm *PubSub) Close() {
	pm.wg.Wait()
}

// AddSubscriber add subscriber
func (pm *PubSub) AddSubscriber(subscribe *pbcom.ReqSubscriber) error {
	if subscribe == nil || subscribe.MsgRouter == nil {
		pm.log.Errorf("addSubscriber input param is null")
		return types.ErrInvalidParam
	}

	// 订阅者订阅数据类型判断
	if _, ok := subName[subscribe.Type]; !ok {
		pm.log.Errorf("addSubscriber input type %d is error", subscribe.Type)
		return types.ErrInvalidParam
	}

	// 订阅方(即本模块需要将消息推送到订阅方)
	mainTopic := subscribe.GetMsgRouter().GetMainTopic()
	subTopic := subscribe.GetMsgRouter().GetSubTopic()
	eventTy := subscribe.GetMsgRouter().GetEventType()
	if mainTopic == "" || subTopic == "" || eventTy <= 0 {
		pm.log.Errorf("addSubscriber input MainTopic = %s or SubTopic = %s is nul or eventTy = %d ", mainTopic, subTopic, eventTy)
		return types.ErrInvalidParam
	}

	//如果该用户已经注册了订阅请求，则只是确认是否需用重新启动，否则就直接返回
	if subscribeInDB, ok := pm.hasSubscriberExist(subscribe); ok {
		if subscribeInDB.Type != subscribe.Type {
			return ErrNotAllowSubscribeType
		}
		return nil
	}

	return pm.storeSubscriber(subscribe)
}

func (pm *PubSub) hasSubscriberExist(sub *pbcom.ReqSubscriber) (*pbcom.ReqSubscriber, bool) {
	pm.mut.RLock()
	defer pm.mut.RUnlock()
	id := genSubscriberID(sub)
	if v, ok := pm.systemSubs[id]; ok {
		return v.subscribe, true
	}
	if v, ok := pm.normalSubs[id]; ok {
		return v.subscribe, true
	}
	pdb := NewSubscriber(pm.ledgerDB)
	req, err := pdb.GetSubscriber(id)
	return req, err == nil
}

func (pm *PubSub) subscriberCount() int64 {
	lst := pm.ledgerDB.NewList()
	return lst.PrefixCount([]byte(dbm.NamespaceSubscriber))
}

//向数据库添加交易回执订阅信息
func (pm *PubSub) storeSubscriber(sub *pbcom.ReqSubscriber) error {
	pm.mut.Lock()
	id := genSubscriberID(sub)
	sysd := sub.GetSystem()
	if sysd != nil {
		pm.systemSubs[id] = &wrapSub{subscribe: sub}
	} else {
		pm.normalSubs[id] = &wrapSub{subscribe: sub}
	}
	pm.mut.Unlock()
	kv, err := SubscriberKV(sub)
	if err != nil {
		pm.log.Error("SubscriberKV err %v ", err)
		return err
	}
	return pm.ledgerDB.Set(kv.Key, kv.Value)
}

// 删除数据库存储的订阅
func (pm *PubSub) unstoreSubscriber(sub *pbcom.ReqSubscriber) error {
	id := genSubscriberID(sub)
	pm.mut.Lock()
	if sub.GetSystem() != nil {
		delete(pm.systemSubs, id)
	} else {
		delete(pm.normalSubs, id)
	}
	pm.mut.Unlock()
	kv, err := SubscriberKV(sub)
	if err != nil {
		pm.log.Error("SubscriberKV err %v ", err)
		return err
	}
	return pm.ledgerDB.Delete(kv.Key)
}

// DelSubscriber ...
func (pm *PubSub) DelSubscriber(subscribe *pbcom.ReqSubscriber) {
	if subscribe == nil || subscribe.MsgRouter == nil {
		pm.log.Errorf("delSubscriber input param is null")
		return
	}
	pm.unstoreSubscriber(subscribe)
}

// JoinReqRsp 合并请求/反馈
type JoinReqRsp struct {
	req *pbcom.ReqSubscriber
	rsp *pbcom.RspSubscriber
}

// PubNormal ...
func (pm *PubSub) PubNormal(eb *ledger.ExecutedBlock, add bool) {
	if len(pm.normalSubs) == 0 {
		return
	}
	for ty := range subName {
		if ty == SubSysParam { // 将系统参数的subname过滤掉
			continue
		}
		joins := pm.getNormalTypeSubscriber(ty, eb, add)
		for _, join := range joins {
			pm.wg.Add(1)
			go func(sub *pbcom.ReqSubscriber, rsp *pbcom.RspSubscriber) {
				defer pm.wg.Done()
				pm.topicPublish(sub, rsp)
			}(join.req, join.rsp)
		}
	}
}

func (pm *PubSub) topicPublish(sub *pbcom.ReqSubscriber, rsp *pbcom.RspSubscriber) error {
	topicClient := pm.topicClient
	mainTopic := sub.GetMsgRouter().GetMainTopic()
	subTopic := sub.GetMsgRouter().GetSubTopic()
	eventTy := sub.GetMsgRouter().GetEventType()
	msg := topicClient.NewMessage(mainTopic, subTopic, eventTy, rsp)
	err := topicClient.Send(msg, false)
	if err != nil {
		pm.log.Errorf("send sys param to %s, error is %s", mainTopic, err.Error())
		return err
	}
	return nil
}

func (pm *PubSub) getNormalTypeSubscriber(ty int32, eb *ledger.ExecutedBlock, add bool) []*JoinReqRsp {
	pm.mut.Lock()
	var subs []*wrapSub
	for _, sub := range pm.normalSubs {
		if sub.subscribe.Type == ty {
			subs = append(subs, sub)
		}
	}
	pm.mut.Unlock()
	if len(subs) == 0 {
		return nil
	}
	value, err := genSubscribeNormalData(ty, eb)
	if err != nil {
		pm.log.Warnf("genSubscribeNormalData fail err %v", err)
		return nil
	}
	var joins []*JoinReqRsp
	for _, v := range subs {
		suber := v.subscribe
		rsp := &pbcom.RspSubscriber{Type: suber.Type}
		rsp.Content = &pbcom.RspSubscriber_Normal{Normal: &pbcom.NormalData{Add: add, Data: value}}
		join := &JoinReqRsp{req: suber, rsp: rsp}
		joins = append(joins, join)
	}
	return joins
}

func genSubscribeNormalData(ty int32, eb *ledger.ExecutedBlock) ([]byte, error) {
	if ty == SubBlock {
		value, err := proto.Marshal(eb.Block)
		if err != nil {
			return nil, err
		}
		return value, nil
	} else if ty == SubBlockHeader {
		value, err := proto.Marshal(eb.GetBlock().GetHeader())
		if err != nil {
			return nil, err
		}
		return value, nil
	} else if ty == SubBlockExecResult {
		value, err := proto.Marshal(eb.GetBlockExecResult())
		if err != nil {
			return nil, err
		}
		return value, nil
	}
	return nil, ErrNotAllowSubscribeType
}

// PubSystem 过滤系统参数
// TODO 后面可以加入bloom过滤器进行高效筛选
// TODO 后面需要调整一个生效高度，只有生效高度之后的才发布给订阅方
func (pm *PubSub) PubSystem(eb *ledger.ExecutedBlock, add bool) (map[string][]byte, error) {
	if len(pm.systemSubs) == 0 {
		return nil, nil
	}
	// 1.先对整个区块中的所有系统事件进行过滤
	sysData := make(map[string][]byte)
	for i, rec := range eb.GetBlockExecResult().GetTransactionExecResults() {
		if rec.Receipt.TxStatus == ledger.TransactionReceipt_SUCCESSFUL &&
			strings.Contains(eb.Block.Transactions[i].Ty, types.System) { // 防止做恶，判断是否是系统合约
			for _, ev := range rec.Receipt.Events {
				if strings.Contains(ev.Name, types.System) {
					sysData[ev.Name] = ev.Data
				}
			}
		}
	}
	if len(sysData) == 0 {
		return sysData, nil
	}

	// 2.据当前订阅系统事件对事件进行同步发送
	err := pm.PubSystemData(int64(eb.GetBlockExecResult().GetBlockNumber()), sysData, add)

	// 3.将系统参数返回给上层
	return sysData, err
}

// PubSystemData ...
func (pm *PubSub) PubSystemData(blockNum int64, sysData map[string][]byte, add bool) error {
	wg := &sync.WaitGroup{}
	var fail int32
	for _, p := range pm.systemSubs {
		sub := p.subscribe
		rsp := filterSystemSubscriber(sub, sysData, add, blockNum)
		if rsp != nil {
			wg.Add(1)
			go func(sub *pbcom.ReqSubscriber, rsp *pbcom.RspSubscriber) {
				defer wg.Done()
				err := pm.workPublish(sub, rsp)
				if err != nil {
					atomic.StoreInt32(&fail, 1)
				}
			}(sub, rsp)
		}
	}
	wg.Wait()
	var err error
	if fail != 0 {
		err = errors.New("publish subscriber fail")
	}
	return err
}

func (pm *PubSub) workPublish(sub *pbcom.ReqSubscriber, rsp *pbcom.RspSubscriber) error {
	workClient := pm.workClient
	mainTopic := sub.GetMsgRouter().GetMainTopic()
	subTopic := sub.GetMsgRouter().GetSubTopic()
	eventTy := sub.GetMsgRouter().GetEventType()
	msg := workClient.NewMessage(mainTopic, subTopic, eventTy, rsp)
	err := workClient.Send(msg, true)
	if err != nil {
		pm.log.Errorf("send sys param to %s, error is %s", mainTopic, err.Error())
		return err
	}
	res, err := workClient.WaitTimeout(msg, 10*time.Second)
	if err != nil {
		pm.log.Errorf("WaitTimeout sys param to %s , error is %s", mainTopic, err.Error())
		return err
	}
	if res.Data != nil {
		v := res.Data.(*pbcom.Reply)
		if v != nil && !v.IsOk {
			pm.log.Errorf(" reply is not ok")
			return err
		}
	}
	return nil
}

func filterSystemSubscriber(suber *pbcom.ReqSubscriber, sysData map[string][]byte, add bool, blockNum int64) *pbcom.RspSubscriber {
	events := getSubscribeSystemEvent(suber)
	var evs []*pbcom.EventData
	for _, e := range events {
		if v, ok := sysData[e]; ok {
			evs = append(evs, &pbcom.EventData{Event: e, Data: v})
		}
	}
	if len(evs) == 0 {
		return nil
	}
	rsp := &pbcom.RspSubscriber{Type: suber.Type}
	rsp.Content = &pbcom.RspSubscriber_System{System: &pbcom.SystemData{Add: add, BlockNum: blockNum, Datas: evs}}
	return rsp
}

func getSubscribeSystemEvent(suber *pbcom.ReqSubscriber) []string {
	var strs []string
	if suber == nil {
		return strs
	}

	for _, params := range suber.GetSystem().GetParams() {
		strs = append(strs, params.Event...)
	}
	return strs
}

func genSubscriberID(sub *pbcom.ReqSubscriber) string {
	if sub == nil || sub.MsgRouter == nil {
		return ""
	}
	return hex.EncodeToString(sub.MsgRouter.Hash())
}
